部将|放入_如何扩展HiveMetastoreThriftRPC服务接口

篇首语：本文由编程笔记#小编为大家整理，主要介绍了如何扩展Hive Metastore Thrift RPC服务接口相关的知识，希望对你有一定的参考价值。

Hive 在经历十几年的发展&＃xff0c;已经获得广泛应用&＃xff0c;随着版本升级&＃xff0c;不同版本之间的协议接口会发生一些变化&＃xff0c;尽管HMS在尽量保持向前兼容&＃xff0c;但在大版本变更时&＃xff0c;仍然不能保证完全兼容&＃xff0c;比如HMS2到HMS3&＃xff0c;有关索引&＃xff08;Index&＃xff09;相关的接口变成了约束&＃xff08;constraint&＃xff09;&＃xff0c;比如下图中左边是HMS3接口定义&＃xff0c;右边是HMS2接口定义&＃xff1a;

这样当使用Hive 2的Client去访问HMS 3 的服务时候&＃xff0c;就会报无效方法&＃xff08;Invalid Method&＃xff09;异常&＃xff0c;此时可以通过一个Hive Metastore代理服务来解决。

在具体介绍Hive Metastore代理服务前&＃xff0c;需要了解下Hive Metastore服务处理流程&＃xff1a;

HMSHandler baseHandler &＃61; new HiveMetaStore.HMSHandler("metaserver", conf,false);
IHMSHandler handler &＃61; newRetryingHMSHandler(baseHandler, conf);
TProcessor processor &＃61; new ThriftHiveMetastore.Processor<>(handler));
final TThreadPoolServer.Args serverArgs &＃61; new TThreadPoolServer
.Args(serverTransport)
.transportFactory(new TTransportFactory())
.protocolFactory(new TBinaryProtocol.Factory())
.processor(processor);
TServer server &＃61; new TThreadPoolServer(serverArgs);
server.setServerEventHandler(getServerEventHandler());
server.serve();

上述示例代码启动了Hive Metastore的服务&＃xff0c;其中两个重要的组件&＃xff1a;

• TThreadPoolServer&＃xff1a;负责请求的监听和派遣

• TProcessor&＃xff1a;负责执行各RPC接口的实现逻辑

TThreadPoolServer功能介绍

TThreadPoolServer主要负责请求的监听和派遣&＃xff0c;当接收到新的请求时&＃xff0c;在内部将每个client的请求放入Worker线程池&＃xff0c;然后继续下一个请求的监听&＃xff0c;而工作线程负责一对一处理每个client的请求&＃xff0c;直至处理完毕&＃xff1a;

#TThreadPoolServer.class
public class TThreadPoolServer extends TServer
//private ExecutorService executorService_;

public void serve()
...
while(!this.stopped_)
TTransport client &＃61; this.serverTransport_.accept();
TThreadPoolServer.WorkerProcess wp &＃61; new TThreadPoolServer.WorkerProcess(client);
this.executorService_.execute(wp);

...
this.executorService_.shutdown();

...
private class WorkerProcess implements Runnable
//private TTransport client_;
public void run()
do
if (eventHandler !&＃61; null)
eventHandler.processContext(connectionContext, inputTransport, outputTransport);

while(!TThreadPoolServer.this.stopped_ && processor.process(inputProtocol, outputProtocol));
if (this.client_.isOpen())
this.client_.close();





TProcessor功能介绍

上面介绍了WorkerProcess的工作功能&＃xff0c;TThreadPoolServer将每个客户端连接整体扔给TProcessor&＃xff0c;没有介绍在其通过processor.process处理时候&＃xff0c;如何定位每个不同的方法&＃xff0c;实际上区分不同方法调用的逻辑同时也是TProcessor负责的&＃xff0c;以Processor为例&＃xff0c;定义了所支持的接口映射&＃xff1a;

#ThriftHiveMetastore.class
public static class Processor extends com.facebook.fb303.FacebookService.Processor implements TProcessor
private static final Logger LOGGER &＃61; LoggerFactory.getLogger(ThriftHiveMetastore.Processor.class.getName());
public Processor(I iface)
super(iface, getProcessMap(new HashMap()));

//Note:processMap
protected Processor(I iface, Map> processMap)
super(iface, getProcessMap(processMap));

private static Map> getProcessMap(Map> processMap)
processMap.put("create_catalog", new ThriftHiveMetastore.Processor.create_catalog());
processMap.put("alter_catalog", new ThriftHiveMetastore.Processor.alter_catalog());
processMap.put("get_catalog", new ThriftHiveMetastore.Processor.get_catalog());
processMap.put("get_catalogs", new ThriftHiveMetastore.Processor.get_catalogs());
processMap.put("drop_catalog", new ThriftHiveMetastore.Processor.drop_catalog());
processMap.put("create_database", new ThriftHiveMetastore.Processor.create_database());
processMap.put("get_database", new ThriftHiveMetastore.Processor.get_database());
processMap.put("drop_database", new ThriftHiveMetastore.Processor.drop_database());
...



也就是凡是登记在processMap的方法才能被识别&＃xff0c;否则都会报异常&＃xff0c;这个逻辑可以在Processor的父类TBaseProcessor中得知&＃xff1a;